package com.zetyun.tiger.connector.op.v1;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Sets;
import com.zetyun.rt.meta.annotation.ActionMeta;
import com.zetyun.rt.sdk.action.MapAction;
import com.zetyun.rt.sdk.model.RtEvent;
import com.zetyun.rt.sdk.operator.OperatorContext;
import com.zetyun.tiger.connector.cdc.SimpleRedisContainer;
import com.zetyun.tiger.connector.util.SpecialDateFormatUtil;
import lombok.SneakyThrows;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisConfigBase;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisSentinelConfig;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisSentinelPool;

import java.util.Objects;
import java.util.Set;

@ActionMeta(
        id = "tiger.redis-op",
        tags = {"redis"},
        category = "map",
        name = "RedisOp",
        description = "Redis操作算子",
        parameterClass = RedisOpParameter.class,
        status = ActionMeta.Status.PUBLIC
)
public class RedisOpAction extends MapAction {

    private SimpleRedisContainer rc;
    private RedisOpParameter parameters;
    private ObjectMapper om;

    private final Integer ttl = 60 * 60 * 24;  // 一天

    @Override
    public void init(OperatorContext context) {
        super.init(context);

        om = new ObjectMapper();
        parameters = context.getParameters();
        FlinkJedisSentinelConfig conf =
                new FlinkJedisSentinelConfig.Builder()
                        .setMasterName(parameters.getMasterName())
                        .setSentinels(parameters.getSentinels())
                        .build();
        rc = build(conf);
        try {
            rc.open();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public RtEvent apply(RtEvent in) throws Exception {
        // hash
        // key:客户经理号
        //          field: 客户号
        //          value: Set<内部账号>
        String dayId = in.getField(parameters.getDayId(), String.class);
        String conntrNo = in.getField(parameters.getConntrNo(), String.class);
        String custNo = in.getField(parameters.getCustNo(), String.class);
        String acctNo = in.getField(parameters.getAcctNo(), String.class);
        String opacType = in.getField(parameters.getOpacType(), String.class);

        String acctNoCacheName = "dim_acctno".concat("_").concat(conntrNo);
        String dayIdCacheName = "dim_"+dayId.concat("_").concat(conntrNo);

        Set<String> allAcctNos = strToSet(rc.hget(acctNoCacheName, custNo));


        if (opacType.equals("01")) {
            // 开户
            if (allAcctNos.size() == 0) {
                // 没有任何账号存在，确实为开户数据，开户数据保存到当日缓存中
                rc.hset(dayIdCacheName, custNo, "1", ttl);   // 设置一天的过期时间

            }
            allAcctNos.add(acctNo);
            rc.hset(acctNoCacheName, custNo, setToStr(allAcctNos), null);


        } else {
            // 销户
            if (allAcctNos.size() == 1 && allAcctNos.contains(custNo)) {
                // 表明账号信息里面存的是当前销户账户，并且只有一条数据，则为销户数据
                rc.hdel(dayIdCacheName, custNo);
            }
            allAcctNos.remove(custNo);
            rc.hset(acctNoCacheName, custNo, setToStr(allAcctNos), null);
        }

        // 获取当前客户经理的今日客户数
        Long customerCnt = rc.hlen(dayIdCacheName);

        // 获取当前处理时间
        long tms = System.currentTimeMillis();
        String ymdHms = SpecialDateFormatUtil.toYmdHms(tms);
        in.setValue("RT_CUST_CNT", customerCnt);
        in.setValue("RT_PROC_TMS", tms);
        in.setValue("RT_PROC_TIME", ymdHms);

        return in;
    }


    @Override
    public void close() throws Exception {
        super.close();
        rc.close();
    }

    @SneakyThrows
    private Set<String> strToSet(String jsonStr) {
        if (jsonStr == null || jsonStr.isEmpty()) {
            return Sets.newHashSet();
        }
        return om.readValue(jsonStr, new TypeReference<Set<String>>() {
        });
    }

    @SneakyThrows
    private String setToStr(Set<String> set) {
        return om.writeValueAsString(set);
    }


    private static SimpleRedisContainer build(FlinkJedisSentinelConfig jedisSentinelConfig) {
        Objects.requireNonNull(jedisSentinelConfig, "Redis sentinel config should not be Null");

        GenericObjectPoolConfig genericObjectPoolConfig = getGenericObjectPoolConfig(jedisSentinelConfig);

        JedisSentinelPool jedisSentinelPool = new JedisSentinelPool(jedisSentinelConfig.getMasterName(),
                jedisSentinelConfig.getSentinels(), genericObjectPoolConfig,
                jedisSentinelConfig.getConnectionTimeout(), jedisSentinelConfig.getSoTimeout(),
                jedisSentinelConfig.getPassword(), jedisSentinelConfig.getDatabase());
        return new SimpleRedisContainer(jedisSentinelPool);
    }

    private static GenericObjectPoolConfig getGenericObjectPoolConfig(FlinkJedisConfigBase jedisConfig) {
        GenericObjectPoolConfig genericObjectPoolConfig = jedisConfig.getTestWhileIdle() ? new JedisPoolConfig() : new GenericObjectPoolConfig();
        genericObjectPoolConfig.setMaxIdle(jedisConfig.getMaxIdle());
        genericObjectPoolConfig.setMaxTotal(jedisConfig.getMaxTotal());
        genericObjectPoolConfig.setMinIdle(jedisConfig.getMinIdle());
        genericObjectPoolConfig.setTestOnBorrow(jedisConfig.getTestOnBorrow());
        genericObjectPoolConfig.setTestOnReturn(jedisConfig.getTestOnReturn());

        return genericObjectPoolConfig;
    }
}
